package com.atguigu.realtime.dws.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import com.atguigu.realtime.dws.function.KwSplit;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {

    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(10021, 4, "dws_traffic_source_keyword_page_view_window");
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        createPageInfo(tableEnv, groupId);

        filterKeywords(tableEnv);

        tableEnv.createTemporarySystemFunction("KwSplit", KwSplit.class);
        
        KwSplit(tableEnv);

        Table windowAggTable = getWindowAggTable(tableEnv);

        createDorisSink(tableEnv);

        windowAggTable.insertInto("doris_sink").execute();
    }

    private static void createDorisSink(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE doris_sink (\n" +
                "\tstt STRING,\n" +
                "\tedt STRING,\n" +
                "\tcur_date STRING,\n" +
                "\t`keyword` STRING,\n" +
                "\tkeyword_count bigint\n" +
                ")" + SQLUtil.getDorisSinkSQL(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW));
    }

    private static Table getWindowAggTable(StreamTableEnvironment tableEnv) {
        Table windowAggTable = tableEnv.sqlQuery("select \n" +
                "cast(TUMBLE_START(row_time, INTERVAL'10' SECOND) as STRING) stt,\n" +
                "cast(TUMBLE_END(row_time, INTERVAL'10' SECOND) as STRING) edt,\n" +
                "cast(CURRENT_DATE as STRING) cur_date,\n" +
                "  keyword,\n" +
                " count(*) keyword_count\n" +
                "FROM keyword_table\n" +
                "GROUP BY\n" +
                " TUMBLE(row_time, INTERVAL'10' SECOND),\n" +
                " keyword");
        return windowAggTable;
    }

    private static void KwSplit(StreamTableEnvironment tableEnv) {
        Table keywordTable = tableEnv.sqlQuery("SELECT keywords, keyword, row_time " +
                "FROM keywords_table " +
                "LEFT JOIN LATERAL TABLE(KwSplit(keywords)) ON TRUE");

        tableEnv.createTemporaryView("keyword_table", keywordTable);
    }

    private static void filterKeywords(StreamTableEnvironment tableEnv) {
        Table keywordsTable = tableEnv.sqlQuery("select\n" +
                "\tpage['item'] keywords,\n" +
                "\t`row_time`\n" +
                "from page_info\n" +
                "where page['last_page_id'] = 'search'\n" +
                "and page['item_type'] = 'keyword'\n" +
                "and page['item'] is not null");

        tableEnv.createTemporaryView("keywords_table", keywordsTable);
    }

    private static void createPageInfo(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql("create table page_info(\n" +
                "\t`common`  map<STRING, STRING>,\n" +
                "\t`page`  map<STRING, STRING>,\n" +
                "\t`ts` bigint, \n" +
                "\t`row_time` as TO_TIMESTAMP_LTZ(ts, 3), \n" +
                "\tWATERMARK FOR row_time AS row_time - INTERVAL '1' MINUTE\n" +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_PAGE, groupId));
    }
}
